package backend

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"math"
	"math/big"
	"net"
	"reflect"
	"strings"
	"time"
	"unsafe"

	"../mysql"
	"../utils"
	"encoding/gob"
	"regexp"
	"strconv"
)

const digitsPerInteger int = 9
const DATETIMEF_INT_OFS int64 = 0x8000000000
const TIMEF_OFS int64 = 0x800000000000
const TIMEF_INT_OFS int64 = 0x800000
const (
	jsonbSmallOffsetSize = 2
	jsonbLargeOffsetSize = 4

	jsonbKeyEntrySizeSmall = 2 + jsonbSmallOffsetSize
	jsonbKeyEntrySizeLarge = 2 + jsonbLargeOffsetSize

	jsonbValueEntrySizeSmall = 1 + jsonbSmallOffsetSize
	jsonbValueEntrySizeLarge = 1 + jsonbLargeOffsetSize
)

//proxy <-> mysql server
type Conn struct {
	conn net.Conn

	pkg *mysql.PacketIO

	addr     string
	user     string
	password string
	db       string

	capability uint32

	status uint16

	collation mysql.CollationId
	charset   string
	salt      []byte

	pushTimestamp int64
	pkgErr        error
}

func (c *Conn) Connect(addr string, user string, password string, db string) error {
	c.addr = addr
	c.user = user
	c.password = password
	c.db = db

	//use utf8
	c.collation = mysql.DEFAULT_COLLATION_ID
	c.charset = mysql.DEFAULT_CHARSET

	return c.ReConnect()
}

func (c *Conn) ReConnect() error {
	if c.conn != nil {
		c.conn.Close()
	}

	n := "tcp"
	if strings.Contains(c.addr, "/") {
		n = "unix"
	}

	netConn, err := net.Dial(n, c.addr)
	if err != nil {
		return err
	}

	tcpConn := netConn.(*net.TCPConn)

	//SetNoDelay controls whether the operating system should delay packet transmission
	// in hopes of sending fewer packets (Nagle's algorithm).
	// The default is true (no delay),
	// meaning that data is sent as soon as possible after a Write.
	//I set this option false.
	tcpConn.SetNoDelay(false)
	tcpConn.SetKeepAlive(true)
	c.conn = tcpConn
	c.pkg = mysql.NewPacketIO(tcpConn)

	if err := c.readInitialHandshake(); err != nil {
		c.conn.Close()
		return err
	}

	if err := c.writeAuthHandshake(); err != nil {
		c.conn.Close()

		return err
	}

	if _, err := c.readOK(); err != nil {
		c.conn.Close()

		return err
	}

	//we must always use autocommit
	if !c.IsAutoCommit() {
		if _, err := c.exec("set autocommit = 1"); err != nil {
			c.conn.Close()

			return err
		}
	}

	return nil
}

func (c *Conn) Close() error {
	if c.conn != nil {
		c.conn.Close()
		c.conn = nil
		c.salt = nil
		c.pkgErr = nil
	}

	return nil
}

func (c *Conn) readPacket() ([]byte, error) {
	d, err := c.pkg.ReadPacket()
	c.pkgErr = err
	return d, err
}

func (c *Conn) writePacket(data []byte) error {
	err := c.pkg.WritePacket(data)
	c.pkgErr = err
	return err
}

func (c *Conn) readInitialHandshake() error {
	data, err := c.readPacket()
	if err != nil {
		return err
	}

	if data[0] == mysql.ERR_HEADER {
		return errors.New("read initial handshake error")
	}

	if data[0] < mysql.MinProtocolVersion {
		return fmt.Errorf("invalid protocol version %d, must >= 10", data[0])
	}

	//skip mysql version and connection id
	//mysql version end with 0x00
	//connection id length is 4
	pos := 1 + bytes.IndexByte(data[1:], 0x00) + 1 + 4

	c.salt = append(c.salt, data[pos:pos+8]...)

	//skip filter
	pos += 8 + 1

	//capability lower 2 bytes
	c.capability = uint32(binary.LittleEndian.Uint16(data[pos : pos+2]))

	pos += 2

	if len(data) > pos {
		//skip server charset
		//c.charset = data[pos]
		pos += 1

		c.status = binary.LittleEndian.Uint16(data[pos : pos+2])
		pos += 2

		c.capability = uint32(binary.LittleEndian.Uint16(data[pos:pos+2]))<<16 | c.capability

		pos += 2

		//skip auth data len or [00]
		//skip reserved (all [00])
		pos += 10 + 1

		// The documentation is ambiguous about the length.
		// The official Python library uses the fixed length 12
		// mysql-proxy also use 12
		// which is not documented but seems to work.
		c.salt = append(c.salt, data[pos:pos+12]...)
	}

	return nil
}

func (c *Conn) writeAuthHandshake() error {
	// Adjust client capability flags based on server support
	capability := mysql.CLIENT_PROTOCOL_41 | mysql.CLIENT_SECURE_CONNECTION |
		mysql.CLIENT_LONG_PASSWORD | mysql.CLIENT_TRANSACTIONS | mysql.CLIENT_LONG_FLAG

	capability &= c.capability

	//packet length
	//capbility 4
	//max-packet size 4
	//charset 1
	//reserved all[0] 23
	length := 4 + 4 + 1 + 23

	//username
	length += len(c.user) + 1

	//we only support secure connection
	auth := mysql.CalcPassword(c.salt, []byte(c.password))

	length += 1 + len(auth)

	if len(c.db) > 0 {
		capability |= mysql.CLIENT_CONNECT_WITH_DB

		length += len(c.db) + 1
	}

	c.capability = capability

	data := make([]byte, length+4)

	//capability [32 bit]
	data[4] = byte(capability)
	data[5] = byte(capability >> 8)
	data[6] = byte(capability >> 16)
	data[7] = byte(capability >> 24)

	//MaxPacketSize [32 bit] (none)
	//data[8] = 0x00
	//data[9] = 0x00
	//data[10] = 0x00
	//data[11] = 0x00

	//Charset [1 byte]
	data[12] = byte(c.collation)

	//Filler [23 bytes] (all 0x00)
	pos := 13 + 23

	//User [null terminated string]
	if len(c.user) > 0 {
		pos += copy(data[pos:], c.user)
	}
	//data[pos] = 0x00
	pos++

	// auth [length encoded integer]
	data[pos] = byte(len(auth))
	pos += 1 + copy(data[pos+1:], auth)

	// db [null terminated string]
	if len(c.db) > 0 {
		pos += copy(data[pos:], c.db)
		//data[pos] = 0x00
	}

	return c.writePacket(data)
}

func (c *Conn) writeCommand(command byte) error {
	c.pkg.Sequence = 0

	return c.writePacket([]byte{
		0x01, //1 bytes long
		0x00,
		0x00,
		0x00, //sequence
		command,
	})
}

func (c *Conn) BinlogDump(arg []byte, addr string, user string, pass string) error {
	if err := c.writeCommandBuf(arg[0], arg[1:]); err != nil {
		fmt.Println("WriteCommandBuf failed")
	} else {
		fmt.Println("WriteCommandBuf success")
	}
	var schema string       // global para
	var table string        // global para
	var sql string          // global para
	var ColumnType []byte   // global para
	var ColumnMeta []uint16 // global para

	for {
		data, err := c.readPacket()
		if err != nil {
			return err
		}

		if data[0] == mysql.EOF_HEADER && len(data) < 9 {
			fmt.Println(" ReadEOF Success" + " Length: " + strconv.Itoa(len(data)))
			return nil
		}

		// binlog := &BinLogOp{}
		if data[0] == mysql.OK_HEADER {
			if _, err := c.handleOKPacket(data); err != nil {
				fmt.Println(" ReadOk failed" + " length: " + strconv.Itoa(len(data)))
				return err
			} else {
				// timeLayout := "2021-06-13 15:04:05"
				// timeStamp := data[1:5]
				eventType := data[5]

				// binlog.Time = time.Unix(int64(utils.Byte2Int(timeStamp)), 0).Format(timeLayout)

				switch eventType {
				case mysql.ROTATE_EVENT:
					fmt.Printf("EVENT TYPE: %v\n", "ROTATE_EVENT")
				case mysql.FORMAT_DESCRIPTION_EVENT:
					fmt.Printf("EVENT TYPE: %v\n", "FORMAT_DESCRIPTION_EVENT")
				case mysql.QUERY_EVENT:
					fmt.Printf("EVENT TYPE: %v\n", "QUERY_EVENT")
					pos := 20
					// threadId := data[20:24]   // 4 bytes. The ID of the thread that issued this statement.
					// createTime := data[24:28] // 4 bytes. The time in seconds that the statement took to execute.
					pos += 8
					schemaLen := data[pos : pos+1] // 1 byte. The length of the name of the database which was the default database when the statement was executed.
					// errorCode := data[29:31]  // 2 bytes. The error code
					pos += 3
					staVarLen := data[pos : pos+2] // 2 bytes (not present in v1, v3). The length of the status variable block.
					pos += 2
					//statusVar := data[33 : 33+Byte2Int(staVarLen)]                                    // string   [len=$status_vars_length]
					database := data[pos+utils.Byte2Int(staVarLen) : pos+utils.Byte2Int(staVarLen)+utils.Byte2Int(schemaLen)] // string   [len=$schema_length]
					query := data[pos+utils.Byte2Int(staVarLen)+utils.Byte2Int(schemaLen)+1:]                                 // string   text of the query

					schema = string(database)
					sql = string(query)

					fmt.Printf("SCHEMA: %v\n", schema)
					fmt.Printf("SQL: %v\n", sql)
				case mysql.TABLE_MAP_EVENT:
					fmt.Printf("EVENT TYPE: %v\n", "TABLE_MAP_EVENT")
					pos := 20
					// tableId := data[20:26]
					pos += 6
					pos += 2
					schemaNameLen := data[pos : pos+1]
					pos += 1 // 1 byte  schema name length
					schema = string(data[pos : pos+utils.Byte2Int(schemaNameLen)])
					pos += utils.Byte2Int(schemaNameLen)
					pos += 1 // 0x00 skip 1 byte
					tableNameLen := data[pos : pos+1]
					pos += 1 // 1 byte  table name length
					table = string(data[pos : pos+utils.Byte2Int(tableNameLen)])
					pos += utils.Byte2Int(tableNameLen) // table name
					pos += 1                            // 0x00 skip 1 byte

					columnCount := utils.Byte2Int(data[pos : pos+1])
					pos += 1 // column-count

					colType := data[pos : pos+columnCount]

					pos += columnCount

					// var n int
					var err error
					var metaData []byte
					if metaData, _, _, err = GetColumnMetaArray(data[pos:]); err != nil {
						return err
					}

					colMeta, err := GetMeta(metaData, columnCount, colType)
					if err != nil {
						return err
					}

					ColumnType = colType
					ColumnMeta = colMeta
					schema = string(schema)
					table = string(table)

					fmt.Printf("SCHEMA: %v\n", string(schema))
					fmt.Printf("TABLE: %v\n", string(table))
					fmt.Printf("columnCount: %v\n", columnCount)
					fmt.Println("-----------------")
					for i := 0; i < len(colType); i++ {
						fmt.Printf("ColumnType: %v\n", colType[i])
					}
					fmt.Println("-----------------")
					fmt.Println("&&&&&&&&&&&&&&&&&")
					for i := 0; i < len(colMeta); i++ {
						fmt.Printf("ColumnMeta: %v\n", colMeta[i])
					}
					fmt.Println("&&&&&&&&&&&&&&&&&")
					fmt.Printf("TABLE: %v\n", string(table))
				case mysql.WRITE_ROWS_EVENT_V2, mysql.UPDATE_ROWS_EVENT_V2, mysql.DELETE_ROWS_EVENT_V2:
					switch eventType {
					case mysql.WRITE_ROWS_EVENT_V2:
						fmt.Printf("EVENT TYPE: %v\n", "WRITE_ROWS_EVENT_V2")
					case mysql.UPDATE_ROWS_EVENT_V2:
						fmt.Printf("EVENT TYPE: %v\n", "UPDATE_ROWS_EVENT_V2")
					case mysql.DELETE_ROWS_EVENT_V2:
						fmt.Printf("EVENT TYPE: %v\n", "DELETE_ROWS_EVENT_V2")
					}
					pos := 20
					// tableId := data[pos : pos+6] // 6 bytes Table ID is 6 byte  | flags 2 bytes
					pos += 6
					// flags := data[pos:pos+2]   	// 2 byte
					pos += 2
					extraDataLen := data[pos : pos+2] // 2 bytes length of extra_data (has to be ≥ 2) if version == 2
					pos += utils.Byte2Int(extraDataLen)

					//conn := new(Conn)
					//conn.Connect(addr, user, pass, "")
					//cols := conn.slecClo(schema, table)

					columnLen := data[pos : pos+1] // 1 byte
					fmt.Printf("columnLen: %v\n", columnLen)
					pos += 1
					colPreBitmap1 := data[pos : pos+(utils.Byte2Int(columnLen)+7)/8] // columns-present-bitmap1, length: (num of columns+7)/8
					pos += (utils.Byte2Int(columnLen) + 7) / 8

					var colPreBitmap2 []byte
					if eventType == mysql.UPDATE_ROWS_EVENT_V2 {
						colPreBitmap2 = data[pos : pos+(utils.Byte2Int(columnLen)+7)/8] // columns-present-bitmap2, length: (num of columns+7)/8
						pos += (utils.Byte2Int(columnLen) + 7) / 8
					}

					var rows1 [][]interface{}
					var rows2 [][]interface{}
					for pos < len(data) {
						// repeat rows until event-end
						rows1Re, n1, err := GetRows(rows1, data[pos:], utils.Byte2Int(columnLen), ColumnType, colPreBitmap1, ColumnMeta)
						if err != nil {
							return err
						}
						pos += n1
						for i := 0; i < len(rows1Re); i++ {
							fmt.Printf("ColumnMeta: %v\n", rows1Re[i])
						}

						if len(colPreBitmap2) > 0 {
							rows2Re, n2, err := GetRows(rows2, data[pos:], utils.Byte2Int(columnLen), ColumnType, colPreBitmap2, ColumnMeta)
							if err != nil {
								return err
							}
							pos += n2
							for i := 0; i < len(rows2Re); i++ {
								fmt.Printf("ColumnMeta: %v\n", rows2Re[i])
							}
						}
					}
				case mysql.ROWS_QUERY_LOG_EVENT:
					fmt.Printf("EVENT TYPE: %v\n", "ROWS_QUERY_LOG_EVENT")
				case mysql.XID_EVENT:
					fmt.Printf("EVENT TYPE: %v\n", "XID_EVENT")
				default:
					fmt.Printf("EVENT TYPE: %v\n", "This event will ignored!")
				}
			}
		}
	}
}

type BinLogOp struct {
	Event      string
	Op         string
	Time       string
	Schema     string
	Table      string
	Column     []string
	BeVal      []string
	AfVal      []string
	Ids        []string
	Sql        string
	ColumnType []byte
	ColumnMeta []uint16
}

type ColumnType struct {
	Col       string
	ColType   string
	ColSuffix []string
	//Length  int
	//LengthF int

}

func (c *Conn) slecClo(schema string, table string) []*ColumnType {

	var cols []*ColumnType
	regStr := regexp.MustCompile("[a-zA-Z]+")
	//regSMath := regexp.MustCompile("[\\d]+")

	comStr := "SELECT COLUMN_NAME, COLUMN_TYPE FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s'"
	com := fmt.Sprintf(comStr, schema, table)
	var res *mysql.Result
	res, err := c.exec(com)

	if err != nil {
		fmt.Println("Execute failed ____________________")
	}

	for i := 0; i < len(res.Resultset.Values); i++ {

		clo := new(ColumnType)

		cloStr, err := res.Resultset.GetValue(i, 0)
		if err != nil {
			panic(err)
		}
		clo.Col = cloStr.(string)

		var cloType interface{}
		var data []byte

		result, err := res.Resultset.GetValue(i, 1)
		if err != nil {
			panic(err)
		}
		if result == nil {
			break
		}

		switch v := result.(type) {
		case []byte:
			data = v
		default:
			cloType, _ = getExprValue(res, i, 1)
			var buf bytes.Buffer
			enc := gob.NewEncoder(&buf)
			erre := enc.Encode(cloType)
			if erre != nil {
				panic(erre)
			}
			data = buf.Bytes()
		}

		var tmpType string
		if strings.LastIndex(string(data), "(") != -1 {
			clo.ColSuffix = strings.Split(string(data)[strings.LastIndex(string(data), "(")+1:strings.Index(string(data), ")")], ",")
			for i := 0; i < len(regStr.FindAllString(string(data)[:strings.LastIndex(string(data), "(")], -1)); i++ {
				tmpType += regStr.FindAllString(string(data)[:strings.LastIndex(string(data), "(")], -1)[i]
			}
		} else {
			tmpType = string(data)
		}

		//var tmpS string
		//for i := 0; i < len(regStr.FindAllString(string(data), -1)); i++ {
		//	tmpS += regStr.FindAllString(string(data), -1)[i]
		//}

		//clo.Length, err = strconv.Atoi(regSMath.FindString(string(data)))
		//if err != nil {
		//	panic(err)
		//}
		//
		//if strings.LastIndex(string(data), ",") != -1 {
		//	tmpI, err := strconv.Atoi(regSMath.FindString(string(data)[strings.Index(string(data), ","):]))
		//	if err != nil {
		//		panic(err)
		//	}
		//	clo.LengthF = tmpI
		//}

		clo.ColType = tmpType
		cols = append(cols, clo)
	}

	return cols
}

func getExprValue(rs *mysql.Result, row int, col int) (interface{}, error) {
	var sumf float64
	var sumi int64
	var IsInt bool
	var err error
	var result interface{}

	result, err = rs.GetValue(row, col)
	if err != nil {
		return nil, err
	}
	if result == nil {
		return nil, fmt.Errorf("invalid resultset")
	}

	switch v := result.(type) {
	case int:
		sumi = sumi + int64(v)
		IsInt = true
	case int32:
		sumi = sumi + int64(v)
		IsInt = true
	case int64:
		sumi = sumi + v
		IsInt = true
	case float32:
		sumf = sumf + float64(v)
	case float64:
		sumf = sumf + v
	default:
		return nil, fmt.Errorf("invalid resultset")
	}
	if IsInt {
		return sumi, nil
	} else {
		return sumf, nil
	}
}

func (c *Conn) CheckSumEnable(com string) (bool, error) {
	var res *mysql.Result
	res, err := c.exec(com)
	if err != nil {
		panic(err)
	}

	var n = len(res.Resultset.Values)
	//fmt.Println(n)

	if n == 1 {
		binlogChecksumStr, _ := res.Resultset.GetValue(0, 1)
		if binlogChecksumStr == "CRC32" {
			return true, nil
		} else if binlogChecksumStr == "NONE" {
			return false, nil
		}
	}
	return false, fmt.Errorf("invalid resultset")
}

func (c *Conn) ShowMasterStatus(com string) (string, uint64, error) {
	var res *mysql.Result
	res, err := c.exec(com)
	if err != nil {
		panic(err)
	}

	var n = len(res.Resultset.Values)
	if n == 1 {
		fileStr, _ := res.Resultset.GetValue(0, 0)
		posStr, _ := res.Resultset.GetValue(0, 1)
		fileStrV, okf := fileStr.(string)

		if !okf {
			panic(err)
		}

		posStrV, okp := posStr.(uint64)
		if !okp {
			panic(err)
		}
		return fileStrV, posStrV, nil
	}
	return "", 0, fmt.Errorf("invalid resultset")
}

func (c *Conn) ShowMasterStatusOld(com string) (string, error) {
	var res *mysql.Result
	res, err := c.exec(com)
	if err != nil {
		panic(err)
	}

	var n = len(res.Resultset.Values)

	if n == 1 {
		fileStr, _ := res.Resultset.GetValue(0, 0)
		// posStr, _ := res.Resultset.GetValue(0, 1)
		fmt.Println(fileStr)
		// fmt.Println(posStr)
		fileStrV, ok := fileStr.(string)
		if !ok {
			panic(err)
		}
		//posStrV, ok := posStr.(uint32)
		//if !ok {
		//	fmt.Println(posStrV)
		//}
		return fileStrV, nil
	}
	return "", fmt.Errorf("invalid resultset")
}

func (c *Conn) writeCommandBuf(command byte, arg []byte) error {
	c.pkg.Sequence = 0
	length := len(arg) + 1
	data := make([]byte, length+4)
	data[4] = command
	copy(data[5:], arg)

	return c.writePacket(data)
}

func (c *Conn) writeCommandStr(command byte, arg string) error {
	c.pkg.Sequence = 0

	length := len(arg) + 1

	data := make([]byte, length+4)

	data[4] = command

	copy(data[5:], arg)

	return c.writePacket(data)
}

func (c *Conn) writeCommandUint32(command byte, arg uint32) error {
	c.pkg.Sequence = 0

	return c.writePacket([]byte{
		0x05, //5 bytes long
		0x00,
		0x00,
		0x00, //sequence

		command,

		byte(arg),
		byte(arg >> 8),
		byte(arg >> 16),
		byte(arg >> 24),
	})
}

func (c *Conn) writeCommandStrStr(command byte, arg1 string, arg2 string) error {
	c.pkg.Sequence = 0

	data := make([]byte, 4, 6+len(arg1)+len(arg2))

	data = append(data, command)
	data = append(data, arg1...)
	data = append(data, 0)
	data = append(data, arg2...)

	return c.writePacket(data)
}

func (c *Conn) Ping() error {
	if err := c.writeCommand(mysql.COM_PING); err != nil {
		return err
	}

	if _, err := c.readOK(); err != nil {
		return err
	}

	return nil
}

func (c *Conn) UseDB(dbName string) error {
	if c.db == dbName || len(dbName) == 0 {
		return nil
	}

	if err := c.writeCommandStr(mysql.COM_INIT_DB, dbName); err != nil {
		return err
	}

	if _, err := c.readOK(); err != nil {
		return err
	}

	c.db = dbName
	return nil
}

func (c *Conn) GetDB() string {
	return c.db
}

func (c *Conn) GetAddr() string {
	return c.addr
}

func (c *Conn) Execute(command string, args ...interface{}) (*mysql.Result, error) {
	if len(args) == 0 {
		return c.exec(command)
	} else {
		if s, err := c.Prepare(command); err != nil {
			return nil, err
		} else {
			var r *mysql.Result
			r, err = s.Execute(args...)
			s.Close()
			return r, err
		}
	}
}

func (c *Conn) ClosePrepare(id uint32) error {
	return c.writeCommandUint32(mysql.COM_STMT_CLOSE, id)
}

func (c *Conn) Begin() error {
	_, err := c.exec("begin")
	return err
}

func (c *Conn) Commit() error {
	_, err := c.exec("commit")
	return err
}

func (c *Conn) Rollback() error {
	_, err := c.exec("rollback")
	return err
}

func (c *Conn) SetAutoCommit(n uint8) error {
	if n == 0 {
		if _, err := c.exec("set autocommit = 0"); err != nil {
			c.conn.Close()

			return err
		}
	} else {
		if _, err := c.exec("set autocommit = 1"); err != nil {
			c.conn.Close()

			return err
		}
	}
	return nil
}

func (c *Conn) SetCharset(charset string, collation mysql.CollationId) error {
	charset = strings.Trim(charset, "\"'`")

	if collation == 0 {
		collation = mysql.CollationNames[mysql.Charsets[charset]]
	}

	if c.charset == charset && c.collation == collation {
		return nil
	}

	_, ok := mysql.CharsetIds[charset]
	if !ok {
		return fmt.Errorf("invalid charset %s", charset)
	}

	_, ok = mysql.Collations[collation]
	if !ok {
		return fmt.Errorf("invalid collation %s", collation)
	}

	if _, err := c.exec(fmt.Sprintf("SET NAMES %s COLLATE %s", charset, mysql.Collations[collation])); err != nil {
		return err
	} else {
		c.collation = collation
		c.charset = charset
		return nil
	}
}

func (c *Conn) FieldList(table string, wildcard string) ([]*mysql.Field, error) {
	if err := c.writeCommandStrStr(mysql.COM_FIELD_LIST, table, wildcard); err != nil {
		return nil, err
	}

	data, err := c.readPacket()
	if err != nil {
		return nil, err
	}

	fs := make([]*mysql.Field, 0, 4)
	var f *mysql.Field
	if data[0] == mysql.ERR_HEADER {
		return nil, c.handleErrorPacket(data)
	} else {
		for {
			if data, err = c.readPacket(); err != nil {
				return nil, err
			}

			// EOF Packet
			if c.isEOFPacket(data) {
				return fs, nil
			}

			if f, err = mysql.FieldData(data).Parse(); err != nil {
				return nil, err
			}
			fs = append(fs, f)
		}
	}
	return nil, fmt.Errorf("field list error")
}

func (c *Conn) exec(query string) (*mysql.Result, error) {
	if err := c.writeCommandStr(mysql.COM_QUERY, query); err != nil {
		return nil, err
	}

	return c.readResult(false)
}

func (c *Conn) readResultset(data []byte, binary bool) (*mysql.Result, error) {
	result := &mysql.Result{
		Status:       0,
		InsertId:     0,
		AffectedRows: 0,

		Resultset: &mysql.Resultset{},
	}

	// column count
	count, _, n := mysql.LengthEncodedInt(data)

	if n-len(data) != 0 {
		return nil, mysql.ErrMalformPacket
	}

	result.Fields = make([]*mysql.Field, count)
	result.FieldNames = make(map[string]int, count)

	if err := c.readResultColumns(result); err != nil {
		return nil, err
	}

	if err := c.readResultRows(result, binary); err != nil {
		return nil, err
	}

	return result, nil
}

func (c *Conn) readResultColumns(result *mysql.Result) (err error) {
	var i int = 0
	var data []byte

	for {
		data, err = c.readPacket()
		if err != nil {
			return
		}

		// EOF Packet
		if c.isEOFPacket(data) {
			if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
				//result.Warnings = binary.LittleEndian.Uint16(data[1:])
				//todo add strict_mode, warning will be treat as error
				result.Status = binary.LittleEndian.Uint16(data[3:])
				c.status = result.Status
			}

			if i != len(result.Fields) {
				err = mysql.ErrMalformPacket
			}

			return
		}

		result.Fields[i], err = mysql.FieldData(data).Parse()
		if err != nil {
			return
		}

		result.FieldNames[string(result.Fields[i].Name)] = i

		i++
	}
}

func (c *Conn) readResultRows(result *mysql.Result, isBinary bool) (err error) {
	var data []byte

	for {
		data, err = c.readPacket()

		if err != nil {
			return
		}

		// EOF Packet
		if c.isEOFPacket(data) {
			if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
				//result.Warnings = binary.LittleEndian.Uint16(data[1:])
				//todo add strict_mode, warning will be treat as error
				result.Status = binary.LittleEndian.Uint16(data[3:])
				c.status = result.Status
			}

			break
		}

		result.RowDatas = append(result.RowDatas, data)
	}

	result.Values = make([][]interface{}, len(result.RowDatas))

	for i := range result.Values {
		result.Values[i], err = result.RowDatas[i].Parse(result.Fields, isBinary)

		if err != nil {
			return err
		}
	}

	return nil
}

func (c *Conn) readUntilEOF() (err error) {
	var data []byte

	for {
		data, err = c.readPacket()

		if err != nil {
			return
		}

		// EOF Packet
		if c.isEOFPacket(data) {
			return
		}
	}
	return
}

func (c *Conn) isEOFPacket(data []byte) bool {
	return data[0] == mysql.EOF_HEADER && len(data) <= 5
}

func (c *Conn) handleOKPacket(data []byte) (*mysql.Result, error) {
	var n int
	var pos int = 1

	r := new(mysql.Result)

	r.AffectedRows, _, n = mysql.LengthEncodedInt(data[pos:])
	pos += n
	r.InsertId, _, n = mysql.LengthEncodedInt(data[pos:])
	pos += n

	if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
		r.Status = binary.LittleEndian.Uint16(data[pos:])
		c.status = r.Status
		pos += 2

		//todo:strict_mode, check warnings as error
		//Warnings := binary.LittleEndian.Uint16(data[pos:])
		//pos += 2
	} else if c.capability&mysql.CLIENT_TRANSACTIONS > 0 {
		r.Status = binary.LittleEndian.Uint16(data[pos:])
		c.status = r.Status
		pos += 2
	}

	//info
	return r, nil
}

func (c *Conn) handleErrorPacket(data []byte) error {
	e := new(mysql.SqlError)

	var pos int = 1

	e.Code = binary.LittleEndian.Uint16(data[pos:])
	pos += 2

	if c.capability&mysql.CLIENT_PROTOCOL_41 > 0 {
		//skip '#'
		pos++
		e.State = string(data[pos : pos+5])
		pos += 5
	}

	e.Message = string(data[pos:])

	return e
}

func (c *Conn) readOK() (*mysql.Result, error) {
	data, err := c.readPacket()
	if err != nil {
		return nil, err
	}

	if data[0] == mysql.OK_HEADER {
		return c.handleOKPacket(data)
	} else if data[0] == mysql.ERR_HEADER {
		return nil, c.handleErrorPacket(data)
	} else {
		return nil, errors.New("invalid ok packet")
	}
}

func (c *Conn) readResult(binary bool) (*mysql.Result, error) {
	data, err := c.readPacket()
	if err != nil {
		return nil, err
	}

	if data[0] == mysql.OK_HEADER {
		return c.handleOKPacket(data)
	} else if data[0] == mysql.ERR_HEADER {
		return nil, c.handleErrorPacket(data)
	} else if data[0] == mysql.LocalInFile_HEADER {
		return nil, mysql.ErrMalformPacket
	}

	return c.readResultset(data, binary)
}

func (c *Conn) IsAutoCommit() bool {
	return c.status&mysql.SERVER_STATUS_AUTOCOMMIT > 0
}

func (c *Conn) IsInTransaction() bool {
	return c.status&mysql.SERVER_STATUS_IN_TRANS > 0
}

func (c *Conn) GetCharset() string {
	return c.charset
}

func GetRows(rows [][]interface{}, data []byte, colCount int, columnsTypes []byte, bitmap []byte, colMeta []uint16) ([][]interface{}, int, error) {
	row := make([]interface{}, colCount)

	pos := 0

	count := 0
	for i := 0; i < colCount; i++ {
		if isBitSet(bitmap, i) {
			count++
		}
	}
	count = (count + 7) / 8

	nullBitmap := data[pos : pos+count]
	pos += count

	nullbitIndex := 0

	var n int
	var err error
	for i := 0; i < colCount; i++ {
		if !isBitSet(bitmap, i) {
			continue
		}

		isNull := (uint32(nullBitmap[nullbitIndex/8]) >> uint32(nullbitIndex%8)) & 0x01
		nullbitIndex++

		if isNull > 0 {
			row[i] = nil
			continue
		}

		row[i], n, err = GetValue(data[pos:], columnsTypes[i], colMeta[i])

		if err != nil {
			return nil, 0, err
		}
		pos += n
	}

	rows = append(rows, row)
	return rows, pos, nil
}

func GetValue(data []byte, tp byte, meta uint16) (v interface{}, n int, err error) {
	var length int = 0

	if tp == mysql.MYSQL_TYPE_STRING {
		if meta >= 256 {
			b0 := uint8(meta >> 8)
			b1 := uint8(meta & 0xFF)

			if b0&0x30 != 0x30 {
				length = int(uint16(b1) | (uint16((b0&0x30)^0x30) << 4))
				tp = byte(b0 | 0x30)
			} else {
				length = int(meta & 0xFF)
				tp = b0
			}
		} else {
			length = int(meta)
		}
	}

	switch tp {
	case mysql.MYSQL_TYPE_NULL:
		return nil, 0, nil
	case mysql.MYSQL_TYPE_LONG:
		n = 4
		// v = ParseBinaryInt32(data)
		v = int32(binary.LittleEndian.Uint32(data))
	case mysql.MYSQL_TYPE_TINY:
		n = 1
		// v = ParseBinaryInt8(data)
		v = int8(data[0])
	case mysql.MYSQL_TYPE_SHORT:
		n = 2
		// v = ParseBinaryInt16(data)
		v = int16(binary.LittleEndian.Uint16(data))
	case mysql.MYSQL_TYPE_INT24:
		n = 3
		//v = ParseBinaryInt24(data)

		u32 := uint32(uint32(data[0]) | uint32(data[1])<<8 | uint32(data[2])<<16)
		if u32&0x00800000 != 0 {
			u32 |= 0xFF000000
		}
		v = int32(u32)

	case mysql.MYSQL_TYPE_LONGLONG:
		n = 8
		//v = ParseBinaryInt64(data)
		v = int64(binary.LittleEndian.Uint64(data))
	case mysql.MYSQL_TYPE_NEWDECIMAL:
		prec := uint8(meta >> 8)
		scale := uint8(meta & 0xFF)
		v, n, err = getDecimal(data, int(prec), int(scale), true)
	case mysql.MYSQL_TYPE_FLOAT:
		n = 4
		//v = ParseBinaryFloat32(data)
		v = math.Float32frombits(binary.LittleEndian.Uint32(data))
	case mysql.MYSQL_TYPE_DOUBLE:
		n = 8
		//v = ParseBinaryFloat64(data)
		v = math.Float64frombits(binary.LittleEndian.Uint64(data))
	case mysql.MYSQL_TYPE_BIT:
		nbits := ((meta >> 8) * 8) + (meta & 0xFF)
		n = int(nbits+7) / 8

		//use int64 for bit
		v, err = getBit(data, int(nbits), int(n))
	case mysql.MYSQL_TYPE_TIMESTAMP:
		n = 4
		t := binary.LittleEndian.Uint32(data)
		if t == 0 {
			v = formatZeroTime(0, 0)
		} else {
			v = parseFracTime(fracTime{
				Time:                    time.Unix(int64(t), 0),
				Dec:                     0,
				timestampStringLocation: &time.Location{},
			})
		}
	case mysql.MYSQL_TYPE_TIMESTAMP2:
		v, n, err = getTimestamp2(data, meta, &time.Location{})
		v = parseFracTime(v)
	case mysql.MYSQL_TYPE_DATETIME:
		n = 8
		i64 := binary.LittleEndian.Uint64(data)
		if i64 == 0 {
			v = formatZeroTime(0, 0)
		} else {
			d := i64 / 1000000
			t := i64 % 1000000
			v = parseFracTime(fracTime{
				Time: time.Date(
					int(d/10000),
					time.Month((d%10000)/100),
					int(d%100),
					int(t/10000),
					int((t%10000)/100),
					int(t%100),
					0,
					time.UTC,
				),
				Dec: 0,
			})
		}
	case mysql.MYSQL_TYPE_DATETIME2:
		v, n, err = getDatetime2(data, meta)
		v = parseFracTime(v)
	case mysql.MYSQL_TYPE_TIME:
		n = 3
		i32 := uint32(FixedLengthInt(data[0:3]))
		if i32 == 0 {
			v = "00:00:00"
		} else {
			sign := ""
			if i32 < 0 {
				sign = "-"
			}
			v = fmt.Sprintf("%s%02d:%02d:%02d", sign, i32/10000, (i32%10000)/100, i32%100)
		}
	case mysql.MYSQL_TYPE_TIME2:
		v, n, err = getTime2(data, meta)
	case mysql.MYSQL_TYPE_DATE:
		n = 3
		i32 := uint32(FixedLengthInt(data[0:3]))
		if i32 == 0 {
			v = "0000-00-00"
		} else {
			v = fmt.Sprintf("%04d-%02d-%02d", i32/(16*32), i32/32%16, i32%32)
		}

	case mysql.MYSQL_TYPE_YEAR:
		n = 1
		v = int(data[0]) + 1900
	case mysql.MYSQL_TYPE_ENUM:
		l := meta & 0xFF
		switch l {
		case 1:
			v = int64(data[0])
			n = 1
		case 2:
			v = int64(binary.LittleEndian.Uint16(data))
			n = 2
		default:
			err = fmt.Errorf("Unknown ENUM packlen=%d", l)
		}
	case mysql.MYSQL_TYPE_SET:
		n = int(meta & 0xFF)
		nbits := n * 8

		v, err = littleGetBit(data, nbits, n)
	case mysql.MYSQL_TYPE_BLOB:
		v, n, err = getBlob(data, meta)
	case mysql.MYSQL_TYPE_VARCHAR,
		mysql.MYSQL_TYPE_VAR_STRING:
		length = int(meta)
		v, n = GetString(data, length)
	case mysql.MYSQL_TYPE_STRING:
		v, n = GetString(data, length)
	case mysql.MYSQL_TYPE_JSON:
		length = int(FixedLengthInt(data[0:meta]))
		n = length + int(meta)
		v, err = getJsonBinary(data[meta:n])
	case mysql.MYSQL_TYPE_GEOMETRY:
		// MySQL saves Geometry as Blob in binlog
		// Seem that the binary format is SRID (4 bytes) + WKB, outer can use
		// MySQL GeoFromWKB or others to create the geometry data.
		// Refer https://dev.mysql.com/doc/refman/5.7/en/gis-wkb-functions.html
		// I also find some go libs to handle WKB if possible
		// see https://github.com/twpayne/go-geom or https://github.com/paulmach/go.geo
		v, n, err = getBlob(data, meta)
	default:
		err = fmt.Errorf("unsupport type %d in binlog and don't know how to handle", tp)
	}
	return
}

func isBitSet(bitmap []byte, i int) bool {
	return bitmap[i>>3]&(1<<(uint(i)&7)) > 0
}

// LengthEncodedString returns the string read as a bytes slice, whether the value is NULL,
// the number of bytes read and an error, in case the string is longer than
// the input slice
func GetColumnMetaArray(b []byte) ([]byte, bool, int, error) {
	// Get length
	num, isNull, n := GetColumnMeta(b)
	if num < 1 {
		return b[n:n], isNull, n, nil
	}

	n += int(num)

	// Check data length
	if len(b) >= n {
		return b[n-int(num) : n : n], false, n, nil
	}
	return nil, false, n, io.EOF
}

func GetColumnMeta(b []byte) (num uint64, isNull bool, n int) {
	if len(b) == 0 {
		return 0, true, 1
	}

	switch b[0] {
	// 251: NULL
	case 0xfb:
		return 0, true, 1

		// 252: value of following 2
	case 0xfc:
		return uint64(b[1]) | uint64(b[2])<<8, false, 3

		// 253: value of following 3
	case 0xfd:
		return uint64(b[1]) | uint64(b[2])<<8 | uint64(b[3])<<16, false, 4

		// 254: value of following 8
	case 0xfe:
		return uint64(b[1]) | uint64(b[2])<<8 | uint64(b[3])<<16 |
				uint64(b[4])<<24 | uint64(b[5])<<32 | uint64(b[6])<<40 |
				uint64(b[7])<<48 | uint64(b[8])<<56,
			false, 9
	}

	// 0-250: value of first byte
	return uint64(b[0]), false, 1
}

func GetMeta(data []byte, columnCount int, columnType []byte) ([]uint16, error) {
	pos := 0
	columnMeta := make([]uint16, columnCount)
	for i, t := range columnType {
		switch t {
		case mysql.MYSQL_TYPE_STRING:
			var x uint16 = uint16(data[pos]) << 8 //real type
			x += uint16(data[pos+1])              //pack or field length
			columnMeta[i] = x
			pos += 2
		case mysql.MYSQL_TYPE_NEWDECIMAL:
			var x uint16 = uint16(data[pos]) << 8 //precision
			x += uint16(data[pos+1])              //decimals
			columnMeta[i] = x
			pos += 2
		case mysql.MYSQL_TYPE_VAR_STRING,
			mysql.MYSQL_TYPE_VARCHAR,
			mysql.MYSQL_TYPE_BIT:
			columnMeta[i] = binary.LittleEndian.Uint16(data[pos:])
			pos += 2
		case mysql.MYSQL_TYPE_BLOB,
			mysql.MYSQL_TYPE_DOUBLE,
			mysql.MYSQL_TYPE_FLOAT,
			mysql.MYSQL_TYPE_GEOMETRY,
			mysql.MYSQL_TYPE_JSON:
			columnMeta[i] = uint16(data[pos])
			pos++
		case mysql.MYSQL_TYPE_TIME2,
			mysql.MYSQL_TYPE_DATETIME2,
			mysql.MYSQL_TYPE_TIMESTAMP2:
			columnMeta[i] = uint16(data[pos])
			pos++
		case mysql.MYSQL_TYPE_NEWDATE,
			mysql.MYSQL_TYPE_ENUM,
			mysql.MYSQL_TYPE_SET,
			mysql.MYSQL_TYPE_TINY_BLOB,
			mysql.MYSQL_TYPE_MEDIUM_BLOB,
			mysql.MYSQL_TYPE_LONG_BLOB:
			return nil, errors.New("unsupport type in binlog " + string(t))
		default:
			columnMeta[i] = 0
		}
	}
	return columnMeta, nil
}

// FixedLengthInt: little endian
func FixedLengthInt(buf []byte) uint64 {
	var num uint64 = 0
	for i, b := range buf {
		num |= uint64(b) << (uint(i) * 8)
	}
	return num
}

// BFixedLengthInt: big endian
func BFixedLengthInt(buf []byte) uint64 {
	var num uint64 = 0
	for i, b := range buf {
		num |= uint64(b) << (uint(len(buf)-i-1) * 8)
	}
	return num
}

func GetString(data []byte, length int) (v string, n int) {
	if length < 256 {
		length = int(data[0])

		n = int(length) + 1
		v = String(data[1:n])
	} else {
		length = int(binary.LittleEndian.Uint16(data[0:]))
		n = length + 2
		v = String(data[2:n])
	}

	return
}

func String(b []byte) (s string) {
	pbytes := (*reflect.SliceHeader)(unsafe.Pointer(&b))
	pstring := (*reflect.StringHeader)(unsafe.Pointer(&s))
	pstring.Data = pbytes.Data
	pstring.Len = pbytes.Len
	return
}

func littleGetBit(data []byte, nbits int, length int) (value int64, err error) {
	if nbits > 1 {
		switch length {
		case 1:
			value = int64(data[0])
		case 2:
			value = int64(binary.LittleEndian.Uint16(data))
		case 3:
			value = int64(FixedLengthInt(data[0:3]))
		case 4:
			value = int64(binary.LittleEndian.Uint32(data))
		case 5:
			value = int64(FixedLengthInt(data[0:5]))
		case 6:
			value = int64(FixedLengthInt(data[0:6]))
		case 7:
			value = int64(FixedLengthInt(data[0:7]))
		case 8:
			value = int64(binary.LittleEndian.Uint64(data))
		default:
			err = fmt.Errorf("invalid bit length %d", length)
		}
	} else {
		if length != 1 {
			err = fmt.Errorf("invalid bit length %d", length)
		} else {
			value = int64(data[0])
		}
	}
	return
}

func getDecimal(data []byte, precision int, decimals int, useDecimal bool) (interface{}, int, error) {
	//see python mysql replication and https://github.com/jeremycole/mysql_binlog
	integral := (precision - decimals)
	uncompIntegral := int(integral / digitsPerInteger)
	uncompFractional := int(decimals / digitsPerInteger)
	compIntegral := integral - (uncompIntegral * digitsPerInteger)
	compFractional := decimals - (uncompFractional * digitsPerInteger)

	binSize := uncompIntegral*4 + compressedBytes[compIntegral] +
		uncompFractional*4 + compressedBytes[compFractional]

	buf := make([]byte, binSize)
	copy(buf, data[:binSize])

	//must copy the data for later change
	data = buf

	// Support negative
	// The sign is encoded in the high bit of the the byte
	// But this bit can also be used in the value
	value := uint32(data[0])
	var res bytes.Buffer
	var mask uint32 = 0
	if value&0x80 == 0 {
		mask = uint32((1 << 32) - 1)
		res.WriteString("-")
	}

	//clear sign
	data[0] ^= 0x80

	pos, value := getDecimalDecompressValue(compIntegral, data, uint8(mask))
	res.WriteString(fmt.Sprintf("%d", value))

	for i := 0; i < uncompIntegral; i++ {
		value = binary.BigEndian.Uint32(data[pos:]) ^ mask
		pos += 4
		res.WriteString(fmt.Sprintf("%09d", value))
	}

	res.WriteString(".")

	for i := 0; i < uncompFractional; i++ {
		value = binary.BigEndian.Uint32(data[pos:]) ^ mask
		pos += 4
		res.WriteString(fmt.Sprintf("%09d", value))
	}

	if size, value := getDecimalDecompressValue(compFractional, data[pos:], uint8(mask)); size > 0 {
		res.WriteString(fmt.Sprintf("%0*d", compFractional, value))
		pos += size
	}

	if useDecimal {
		f, err := NewFromString(String(res.Bytes()))
		return f, pos, err
	}

	f, err := strconv.ParseFloat(String(res.Bytes()), 64)
	return f, pos, err
}

var compressedBytes = []int{0, 1, 1, 2, 2, 3, 3, 4, 4, 4}

func getDecimalDecompressValue(compIndx int, data []byte, mask uint8) (size int, value uint32) {
	size = compressedBytes[compIndx]
	databuff := make([]byte, size)
	for i := 0; i < size; i++ {
		databuff[i] = data[i] ^ mask
	}
	value = uint32(BFixedLengthInt(databuff))
	return
}

func NewFromString(value string) (Decimal, error) {
	originalInput := value
	var intString string
	var exp int64

	// Check if number is using scientific notation
	eIndex := strings.IndexAny(value, "Ee")
	if eIndex != -1 {
		expInt, err := strconv.ParseInt(value[eIndex+1:], 10, 32)
		if err != nil {
			if e, ok := err.(*strconv.NumError); ok && e.Err == strconv.ErrRange {
				return Decimal{}, fmt.Errorf("can't convert %s to decimal: fractional part too long", value)
			}
			return Decimal{}, fmt.Errorf("can't convert %s to decimal: exponent is not numeric", value)
		}
		value = value[:eIndex]
		exp = expInt
	}

	pIndex := -1
	vLen := len(value)
	for i := 0; i < vLen; i++ {
		if value[i] == '.' {
			if pIndex > -1 {
				return Decimal{}, fmt.Errorf("can't convert %s to decimal: too many .s", value)
			}
			pIndex = i
		}
	}

	if pIndex == -1 {
		// There is no decimal point, we can just parse the original string as
		// an int
		intString = value
	} else {
		if pIndex+1 < vLen {
			intString = value[:pIndex] + value[pIndex+1:]
		} else {
			intString = value[:pIndex]
		}
		expInt := -len(value[pIndex+1:])
		exp += int64(expInt)
	}

	var dValue *big.Int
	// strconv.ParseInt is faster than new(big.Int).SetString so this is just a shortcut for strings we know won't overflow
	if len(intString) <= 18 {
		parsed64, err := strconv.ParseInt(intString, 10, 64)
		if err != nil {
			return Decimal{}, fmt.Errorf("can't convert %s to decimal", value)
		}
		dValue = big.NewInt(parsed64)
	} else {
		dValue = new(big.Int)
		_, ok := dValue.SetString(intString, 10)
		if !ok {
			return Decimal{}, fmt.Errorf("can't convert %s to decimal", value)
		}
	}

	if exp < math.MinInt32 || exp > math.MaxInt32 {
		// NOTE(vadim): I doubt a string could realistically be this long
		return Decimal{}, fmt.Errorf("can't convert %s to decimal: fractional part too long", originalInput)
	}

	return Decimal{
		value: dValue,
		exp:   int32(exp),
	}, nil
}

type Decimal struct {
	value *big.Int
	exp   int32
}

func New(value int64, exp int32) Decimal {
	return Decimal{
		value: big.NewInt(value),
		exp:   exp,
	}
}

func getBit(data []byte, nbits int, length int) (value int64, err error) {
	if nbits > 1 {
		switch length {
		case 1:
			value = int64(data[0])
		case 2:
			value = int64(binary.BigEndian.Uint16(data))
		case 3:
			value = int64(BFixedLengthInt(data[0:3]))
		case 4:
			value = int64(binary.BigEndian.Uint32(data))
		case 5:
			value = int64(BFixedLengthInt(data[0:5]))
		case 6:
			value = int64(BFixedLengthInt(data[0:6]))
		case 7:
			value = int64(BFixedLengthInt(data[0:7]))
		case 8:
			value = int64(binary.BigEndian.Uint64(data))
		default:
			err = fmt.Errorf("invalid bit length %d", length)
		}
	} else {
		if length != 1 {
			err = fmt.Errorf("invalid bit length %d", length)
		} else {
			value = int64(data[0])
		}
	}
	return
}

func getTimestamp2(data []byte, dec uint16, timestampStringLocation *time.Location) (interface{}, int, error) {
	//get timestamp binary length
	n := int(4 + (dec+1)/2)
	sec := int64(binary.BigEndian.Uint32(data[0:4]))
	usec := int64(0)
	switch dec {
	case 1, 2:
		usec = int64(data[4]) * 10000
	case 3, 4:
		usec = int64(binary.BigEndian.Uint16(data[4:])) * 100
	case 5, 6:
		usec = int64(BFixedLengthInt(data[4:7]))
	}

	if sec == 0 {
		return formatZeroTime(int(usec), int(dec)), n, nil
	}

	return fracTime{
		Time:                    time.Unix(sec, usec*1000),
		Dec:                     int(dec),
		timestampStringLocation: timestampStringLocation,
	}, n, nil
}

func formatZeroTime(frac int, dec int) string {
	if dec == 0 {
		return "0000-00-00 00:00:00"
	}

	s := fmt.Sprintf("0000-00-00 00:00:00.%06d", frac)

	// dec must < 6, if frac is 924000, but dec is 3, we must output 924 here.
	return s[0 : len(s)-(6-dec)]
}

type fracTime struct {
	time.Time

	// Dec must in [0, 6]
	Dec int

	timestampStringLocation *time.Location
}

func getBlob(data []byte, meta uint16) (v []byte, n int, err error) {
	var length int
	switch meta {
	case 1:
		length = int(data[0])
		v = data[1 : 1+length]
		n = length + 1
	case 2:
		length = int(binary.LittleEndian.Uint16(data))
		v = data[2 : 2+length]
		n = length + 2
	case 3:
		length = int(FixedLengthInt(data[0:3]))
		v = data[3 : 3+length]
		n = length + 3
	case 4:
		length = int(binary.LittleEndian.Uint32(data))
		v = data[4 : 4+length]
		n = length + 4
	default:
		err = fmt.Errorf("invalid blob packlen = %d", meta)
	}

	return
}

func getDatetime2(data []byte, dec uint16) (interface{}, int, error) {
	//get datetime binary length
	n := int(5 + (dec+1)/2)

	intPart := int64(BFixedLengthInt(data[0:5])) - DATETIMEF_INT_OFS
	var frac int64 = 0

	switch dec {
	case 1, 2:
		frac = int64(data[5]) * 10000
	case 3, 4:
		frac = int64(binary.BigEndian.Uint16(data[5:7])) * 100
	case 5, 6:
		frac = int64(BFixedLengthInt(data[5:8]))
	}

	if intPart == 0 {
		return formatZeroTime(int(frac), int(dec)), n, nil
	}

	tmp := intPart<<24 + frac
	//handle sign???
	if tmp < 0 {
		tmp = -tmp
	}

	// var secPart int64 = tmp % (1 << 24)
	ymdhms := tmp >> 24

	ymd := ymdhms >> 17
	ym := ymd >> 5
	hms := ymdhms % (1 << 17)

	day := int(ymd % (1 << 5))
	month := int(ym % 13)
	year := int(ym / 13)

	second := int(hms % (1 << 6))
	minute := int((hms >> 6) % (1 << 6))
	hour := int((hms >> 12))

	return fracTime{
		Time: time.Date(year, time.Month(month), day, hour, minute, second, int(frac*1000), time.UTC),
		Dec:  int(dec),
	}, n, nil
}

func getTime2(data []byte, dec uint16) (string, int, error) {
	//time  binary length
	n := int(3 + (dec+1)/2)

	tmp := int64(0)
	intPart := int64(0)
	frac := int64(0)
	switch dec {
	case 1:
	case 2:
		intPart = int64(BFixedLengthInt(data[0:3])) - TIMEF_INT_OFS
		frac = int64(data[3])
		if intPart < 0 && frac > 0 {
			/*
			   Negative values are stored with reverse fractional part order,
			   for binary sort compatibility.

			     Disk value  intpart frac   Time value   Memory value
			     800000.00    0      0      00:00:00.00  0000000000.000000
			     7FFFFF.FF   -1      255   -00:00:00.01  FFFFFFFFFF.FFD8F0
			     7FFFFF.9D   -1      99    -00:00:00.99  FFFFFFFFFF.F0E4D0
			     7FFFFF.00   -1      0     -00:00:01.00  FFFFFFFFFF.000000
			     7FFFFE.FF   -1      255   -00:00:01.01  FFFFFFFFFE.FFD8F0
			     7FFFFE.F6   -2      246   -00:00:01.10  FFFFFFFFFE.FE7960

			     Formula to convert fractional part from disk format
			     (now stored in "frac" variable) to absolute value: "0x100 - frac".
			     To reconstruct in-memory value, we shift
			     to the next integer value and then substruct fractional part.
			*/
			intPart++     /* Shift to the next integer value */
			frac -= 0x100 /* -(0x100 - frac) */
		}
		tmp = intPart<<24 + frac*10000
	case 3:
	case 4:
		intPart = int64(BFixedLengthInt(data[0:3])) - TIMEF_INT_OFS
		frac = int64(binary.BigEndian.Uint16(data[3:5]))
		if intPart < 0 && frac > 0 {
			/*
			   Fix reverse fractional part order: "0x10000 - frac".
			   See comments for FSP=1 and FSP=2 above.
			*/
			intPart++       /* Shift to the next integer value */
			frac -= 0x10000 /* -(0x10000-frac) */
		}
		tmp = intPart<<24 + frac*100

	case 5:
	case 6:
		tmp = int64(BFixedLengthInt(data[0:6])) - TIMEF_OFS
	default:
		intPart = int64(BFixedLengthInt(data[0:3])) - TIMEF_INT_OFS
		tmp = intPart << 24
	}

	if intPart == 0 {
		return "00:00:00", n, nil
	}

	hms := int64(0)
	sign := ""
	if tmp < 0 {
		tmp = -tmp
		sign = "-"
	}

	hms = tmp >> 24

	hour := (hms >> 12) % (1 << 10) /* 10 bits starting at 12th */
	minute := (hms >> 6) % (1 << 6) /* 6 bits starting at 6th   */
	second := hms % (1 << 6)        /* 6 bits starting at 0th   */
	secPart := tmp % (1 << 24)

	if secPart != 0 {
		return fmt.Sprintf("%s%02d:%02d:%02d.%06d", sign, hour, minute, second, secPart), n, nil
	}

	return fmt.Sprintf("%s%02d:%02d:%02d", sign, hour, minute, second), n, nil
}

func parseFracTime(t interface{}) interface{} {
	v, ok := t.(fracTime)
	if !ok {
		return t
	}

	// if !e.parseTime {
	// Don't parse time, return string directly
	return v.String()
	// }

	// return Golang time directly
	// return v.Time
}

// decodeJsonBinary decodes the JSON binary encoding data and returns
// the common JSON encoding data.
func getJsonBinary(data []byte) ([]byte, error) {
	// Sometimes, we can insert a NULL JSON even we set the JSON field as NOT NULL.
	// If we meet this case, we can return an empty slice.
	if len(data) == 0 {
		return []byte{}, nil
	}
	d := jsonBinaryDecoder{useDecimal: true}

	if d.isDataShort(data, 1) {
		return nil, d.err
	}

	v := d.decodeValue(data[0], data[1:])
	if d.err != nil {
		return nil, d.err
	}

	return json.Marshal(v)
}

type jsonBinaryDecoder struct {
	useDecimal bool
	err        error
}

func (d *jsonBinaryDecoder) decodeValue(tp byte, data []byte) interface{} {
	if d.err != nil {
		return nil
	}

	switch tp {
	case mysql.JSONB_SMALL_OBJECT:
		return d.decodeObjectOrArray(data, true, true)
	case mysql.JSONB_LARGE_OBJECT:
		return d.decodeObjectOrArray(data, false, true)
	case mysql.JSONB_SMALL_ARRAY:
		return d.decodeObjectOrArray(data, true, false)
	case mysql.JSONB_LARGE_ARRAY:
		return d.decodeObjectOrArray(data, false, false)
	case mysql.JSONB_LITERAL:
		return d.decodeLiteral(data)
	case mysql.JSONB_INT16:
		return d.decodeInt16(data)
	case mysql.JSONB_UINT16:
		return d.decodeUint16(data)
	case mysql.JSONB_INT32:
		return d.decodeInt32(data)
	case mysql.JSONB_UINT32:
		return d.decodeUint32(data)
	case mysql.JSONB_INT64:
		return d.decodeInt64(data)
	case mysql.JSONB_UINT64:
		return d.decodeUint64(data)
	case mysql.JSONB_DOUBLE:
		return d.decodeDouble(data)
	case mysql.JSONB_STRING:
		return d.decodeString(data)
	case mysql.JSONB_OPAQUE:
		return d.decodeOpaque(data)
	default:
		d.err = errors.New("invalid json type %d" + string(tp))
	}

	return nil
}

func (d *jsonBinaryDecoder) isDataShort(data []byte, expected int) bool {
	if d.err != nil {
		return true
	}

	if len(data) < expected {
		d.err = errors.New("data len %d < expected %d" + string(len(data)) + string(expected))
	}

	return d.err != nil
}

func (d *jsonBinaryDecoder) decodeObjectOrArray(data []byte, isSmall bool, isObject bool) interface{} {
	offsetSize := jsonbGetOffsetSize(isSmall)
	if d.isDataShort(data, 2*offsetSize) {
		return nil
	}

	count := d.decodeCount(data, isSmall)
	size := d.decodeCount(data[offsetSize:], isSmall)

	if d.isDataShort(data, int(size)) {
		return nil
	}

	keyEntrySize := jsonbGetKeyEntrySize(isSmall)
	valueEntrySize := jsonbGetValueEntrySize(isSmall)

	headerSize := 2*offsetSize + count*valueEntrySize

	if isObject {
		headerSize += count * keyEntrySize
	}

	if headerSize > size {
		d.err = errors.New("header size %d > size %d" + string(headerSize) + string(size))
		return nil
	}

	var keys []string
	if isObject {
		keys = make([]string, count)
		for i := 0; i < count; i++ {
			// decode key
			entryOffset := 2*offsetSize + keyEntrySize*i
			keyOffset := d.decodeCount(data[entryOffset:], isSmall)
			keyLength := int(d.decodeUint16(data[entryOffset+offsetSize:]))

			// Key must start after value entry
			if keyOffset < headerSize {
				d.err = errors.New("invalid key offset %d, must > %d" + string(keyOffset) + string(headerSize))
				return nil
			}

			if d.isDataShort(data, keyOffset+keyLength) {
				return nil
			}

			keys[i] = String(data[keyOffset : keyOffset+keyLength])
		}
	}

	if d.err != nil {
		return nil
	}

	values := make([]interface{}, count)
	for i := 0; i < count; i++ {
		// decode value
		entryOffset := 2*offsetSize + valueEntrySize*i
		if isObject {
			entryOffset += keyEntrySize * count
		}

		tp := data[entryOffset]

		if isInlineValue(tp, isSmall) {
			values[i] = d.decodeValue(tp, data[entryOffset+1:entryOffset+valueEntrySize])
			continue
		}

		valueOffset := d.decodeCount(data[entryOffset+1:], isSmall)

		if d.isDataShort(data, valueOffset) {
			return nil
		}

		values[i] = d.decodeValue(tp, data[valueOffset:])
	}

	if d.err != nil {
		return nil
	}

	if !isObject {
		return values
	}

	m := make(map[string]interface{}, count)
	for i := 0; i < count; i++ {
		m[keys[i]] = values[i]
	}

	return m
}

func (d *jsonBinaryDecoder) decodeCount(data []byte, isSmall bool) int {
	if isSmall {
		v := d.decodeUint16(data)
		return int(v)
	}

	return int(d.decodeUint32(data))
}

func (d *jsonBinaryDecoder) decodeUint16(data []byte) uint16 {
	if d.isDataShort(data, 2) {
		return 0
	}

	v := ParseBinaryUint16(data[0:2])
	return v
}

func (d *jsonBinaryDecoder) decodeUint32(data []byte) uint32 {
	if d.isDataShort(data, 4) {
		return 0
	}

	v := ParseBinaryUint32(data[0:4])
	return v
}

func ParseBinaryUint16(data []byte) uint16 {
	return binary.LittleEndian.Uint16(data)
}

func ParseBinaryUint32(data []byte) uint32 {
	return binary.LittleEndian.Uint32(data)
}

func isInlineValue(tp byte, isSmall bool) bool {
	switch tp {
	case mysql.JSONB_INT16, mysql.JSONB_UINT16, mysql.JSONB_LITERAL:
		return true
	case mysql.JSONB_INT32, mysql.JSONB_UINT32:
		return !isSmall
	}

	return false
}

func jsonbGetKeyEntrySize(isSmall bool) int {
	if isSmall {
		return jsonbKeyEntrySizeSmall
	}

	return jsonbKeyEntrySizeLarge
}

func jsonbGetValueEntrySize(isSmall bool) int {
	if isSmall {
		return jsonbValueEntrySizeSmall
	}

	return jsonbValueEntrySizeLarge
}

func jsonbGetOffsetSize(isSmall bool) int {
	if isSmall {
		return jsonbSmallOffsetSize
	}

	return jsonbLargeOffsetSize
}

func (d *jsonBinaryDecoder) decodeLiteral(data []byte) interface{} {
	if d.isDataShort(data, 1) {
		return nil
	}

	tp := data[0]

	switch tp {
	case mysql.JSONB_NULL_LITERAL:
		return nil
	case mysql.JSONB_TRUE_LITERAL:
		return true
	case mysql.JSONB_FALSE_LITERAL:
		return false
	}

	d.err = errors.New("invalid literal %c" + string(tp))

	return nil
}

func (d *jsonBinaryDecoder) decodeInt16(data []byte) int16 {
	if d.isDataShort(data, 2) {
		return 0
	}

	v := ParseBinaryInt16(data[0:2])
	return v
}

func ParseBinaryInt16(data []byte) int16 {
	return int16(binary.LittleEndian.Uint16(data))
}

func (d *jsonBinaryDecoder) decodeInt32(data []byte) int32 {
	if d.isDataShort(data, 4) {
		return 0
	}

	v := ParseBinaryInt32(data[0:4])
	return v
}

func ParseBinaryInt32(data []byte) int32 {
	return int32(binary.LittleEndian.Uint32(data))
}

func (d *jsonBinaryDecoder) decodeInt64(data []byte) int64 {
	if d.isDataShort(data, 8) {
		return 0
	}

	v := ParseBinaryInt64(data[0:8])
	return v
}

func ParseBinaryInt64(data []byte) int64 {
	return int64(binary.LittleEndian.Uint64(data))
}
func (d *jsonBinaryDecoder) decodeUint64(data []byte) uint64 {
	if d.isDataShort(data, 8) {
		return 0
	}

	v := ParseBinaryUint64(data[0:8])
	return v
}

func ParseBinaryUint64(data []byte) uint64 {
	return binary.LittleEndian.Uint64(data)
}

func (d *jsonBinaryDecoder) decodeDouble(data []byte) float64 {
	if d.isDataShort(data, 8) {
		return 0
	}

	v := ParseBinaryFloat64(data[0:8])
	return v
}
func ParseBinaryFloat64(data []byte) float64 {
	return math.Float64frombits(binary.LittleEndian.Uint64(data))
}
func (d *jsonBinaryDecoder) decodeString(data []byte) string {
	if d.err != nil {
		return ""
	}

	l, n := d.decodeVariableLength(data)

	if d.isDataShort(data, l+n) {
		return ""
	}

	data = data[n:]

	v := String(data[0:l])
	return v
}

func (d *jsonBinaryDecoder) decodeVariableLength(data []byte) (int, int) {
	// The max size for variable length is math.MaxUint32, so
	// here we can use 5 bytes to save it.
	maxCount := 5
	if len(data) < maxCount {
		maxCount = len(data)
	}

	pos := 0
	length := uint64(0)
	for ; pos < maxCount; pos++ {
		v := data[pos]
		length |= uint64(v&0x7F) << uint(7*pos)

		if v&0x80 == 0 {
			if length > math.MaxUint32 {
				d.err = errors.New("variable length %d must <= %d" + string(length) + string(int64(math.MaxUint32)))
				return 0, 0
			}

			pos += 1
			// TODO: should consider length overflow int here.
			return int(length), pos
		}
	}

	d.err = errors.New("decode variable length failed")

	return 0, 0
}
func (d *jsonBinaryDecoder) decodeOpaque(data []byte) interface{} {
	if d.isDataShort(data, 1) {
		return nil
	}

	tp := data[0]
	data = data[1:]

	l, n := d.decodeVariableLength(data)

	if d.isDataShort(data, l+n) {
		return nil
	}

	data = data[n : l+n]

	switch tp {
	case mysql.MYSQL_TYPE_NEWDECIMAL:
		return d.decodeDecimal(data)
	case mysql.MYSQL_TYPE_TIME:
		return d.decodeTime(data)
	case mysql.MYSQL_TYPE_DATE, mysql.MYSQL_TYPE_DATETIME, mysql.MYSQL_TYPE_TIMESTAMP:
		return d.decodeDateTime(data)
	default:
		return String(data)
	}

	return nil
}
func (d *jsonBinaryDecoder) decodeDateTime(data []byte) interface{} {
	v := d.decodeInt64(data)
	if v == 0 {
		return "0000-00-00 00:00:00"
	}

	// handle negative?
	if v < 0 {
		v = -v
	}

	intPart := v >> 24
	ymd := intPart >> 17
	ym := ymd >> 5
	hms := intPart % (1 << 17)

	year := ym / 13
	month := ym % 13
	day := ymd % (1 << 5)
	hour := (hms >> 12)
	minute := (hms >> 6) % (1 << 6)
	second := hms % (1 << 6)
	frac := v % (1 << 24)

	return fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d.%06d", year, month, day, hour, minute, second, frac)

}
func (d *jsonBinaryDecoder) decodeDecimal(data []byte) interface{} {
	precision := int(data[0])
	scale := int(data[1])

	v, _, err := getDecimal(data[2:], precision, scale, d.useDecimal)
	d.err = err

	return v
}
func (d *jsonBinaryDecoder) decodeTime(data []byte) interface{} {
	v := d.decodeInt64(data)

	if v == 0 {
		return "00:00:00"
	}

	sign := ""
	if v < 0 {
		sign = "-"
		v = -v
	}

	intPart := v >> 24
	hour := (intPart >> 12) % (1 << 10)
	min := (intPart >> 6) % (1 << 6)
	sec := intPart % (1 << 6)
	frac := v % (1 << 24)

	return fmt.Sprintf("%s%02d:%02d:%02d.%06d", sign, hour, min, sec, frac)
}
